package producer.service;

import com.api.Message;
import com.api.MessageType;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import producer.broker.AsyncBaseQueue;
import producer.broker.RabbitTemplateContainer;
import producer.constant.BrokerMessageConst;
import producer.constant.BrokerMessageStatus;
import producer.entity.BrokerMessage;

import java.util.Date;
import java.util.List;

/**
 * Created with IDEA
 * Author: LuoHengYi
 * Date: 2022/6/18
 * Time: 10:58
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class RabbitBrokerImpl implements RabbitBroker {

    private final RabbitTemplateContainer rabbitTemplateContainer;

    private final MessageStoreService messageStoreService;


    @Override
    public void rapidSend(Message message) {
        message.setMessageType(MessageType.RAPID);
        sendKernel(message);
    }

    /**
     * $sendKernel 发送消息的核心方法,使用异步方式发送消息
     * @param message
     */
    private void sendKernel(Message message) {

        AsyncBaseQueue.submit(() ->{

            // mq 消息发布成功，做 confirm 时回发送回来
            CorrelationData correlationData  = new CorrelationData(
                    String.format("%s#%s",message.getMessageId(),System.currentTimeMillis())
            );
            String topic = message.getTopic();
            String routingKey = message.getRoutingKey();

            //去连接池中更具 msg信息获取一个msg
            RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getRabbitTemplate(message);
            rabbitTemplate.convertAndSend(topic,routingKey,message,correlationData);


            log.info("#RabbitBrokerImpl.sendKernel# send to rabbitmq，MessageId: {}",message.getMessageId());

        });

    }

    @Override
    public void confirmSend(Message message) {
        message.setMessageType(MessageType.CONFIRM);
        sendKernel(message);
    }

    /**
     * 可靠性投递
     * @param message
     */
    @Override
    public void reliantSend(Message message) {

        message.setMessageType(MessageType.RELIANT);

        Date date = new Date();

        BrokerMessage brokerMessage = new BrokerMessage();

        brokerMessage.setMessageId(message.getMessageId());
        brokerMessage.setStatus(BrokerMessageStatus.SENDING.getCode());

        //设置超过多久重新发送，这里是1分钟
        brokerMessage.setNextRetry(DateUtils.addMinutes(date, BrokerMessageConst.TIMEOUT));

        brokerMessage.setCreateTime(date);
        brokerMessage.setUpdateTime(date);
        brokerMessage.setMessage(message);
        //消息入库
        messageStoreService.insert(brokerMessage);

        //发送消息
        sendKernel(message);

    }

    @Override
    public void sendMessages(List<Message> messages) {

    }
}
